/*
 * Copyright (c) 2019.  drakelee.base All rights reserved.
 */

package com.base.components.task.service;

import com.base.components.common.service.task.SysTask;
import com.base.components.common.util.JsonUtils;
import com.base.components.common.util.UUIDUtils;
import com.base.components.database.dao.base.condition.ConditionGroup;
import com.base.components.task.exception.DestroyingException;
import com.base.components.task.exception.TaskExecuteException;
import com.fasterxml.jackson.databind.node.ObjectNode;
import com.google.common.base.Strings;
import com.google.common.util.concurrent.Uninterruptibles;
import org.apache.commons.lang3.StringUtils;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.Pageable;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.util.Assert;

import javax.persistence.EntityManager;
import java.lang.invoke.MethodHandles;
import java.util.Collection;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

/**
 * SysTask Service  JPA
 * @author : code generator
 * @version : 1.0
 * @since : 2017-11-20
 */
public abstract class SysTaskService{

  private static final Logger logger = LoggerFactory.getLogger(MethodHandles.lookup().lookupClass());

  @Autowired
  private ServerClientService serverClientService;

  @Autowired
  private TaskManager taskManager;

  @Autowired
  private PlatformTransactionManager transactionManager;

  public void finishAndPersistence(SysTask task, DateTime currentDate, boolean setDisabled,
                                   boolean doneTask){
    if (task != null) {
      DefaultTransactionDefinition def =
        new DefaultTransactionDefinition(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
      TransactionStatus status = transactionManager.getTransaction(def);
      try {
        task = findForUpdate(task.getId());
        if(task == null){
          return;
        }
        task.setLocking(Boolean.FALSE);
        if (doneTask) {
          task.setDisabled(setDisabled);
          task.setLastTime(currentDate.toDate());
          task.setLastServer(TaskManager.serverHost);
        }
        getEntityManager().merge(task);
        transactionManager.commit(status);
      }catch (Exception e) {
        transactionManager.rollback(status);
        logger.error("完成作业保存失败！可能会导致当前作业的其它服务器线程长时间阻塞！已尝试广播解锁此作业（" + task.getId() + "）！", e);
        serverClientService.sendBroadcast(task.getId(), TaskBroadcastType.UNLOCK, true);
      }
    }
  }

  public SysTask findAndLock(String id){
    SysTask task = null;
    long begin = System.currentTimeMillis();
    while (true) {
      DefaultTransactionDefinition def =
        new DefaultTransactionDefinition(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
      TransactionStatus status = transactionManager.getTransaction(def);
      boolean locking = false;
      try {
        task = findForUpdate(id);
        if(task != null){
          //未锁定，继续执行作业
          if (task.getLocking() != null && !task.getLocking()) {
            TaskManager.addDoingTask(task);
            //设置为锁定
            task.setLocking(Boolean.TRUE);
            getEntityManager().merge(task);
          }
          //已锁定，当前线程睡眠固定时间后继续循环获读取作业实体
          else {
            locking = true;
          }
        }
        transactionManager.commit(status);
      } catch (Exception e) {
        transactionManager.rollback(status);
        if(e instanceof DestroyingException){
          throw e;
        }
        else{
          logger.error("findAndLock", e);
        }
      }
      if (task == null) {
        return null;
      }
      if (!locking) {
        return task;
      }
      else {
        Uninterruptibles.sleepUninterruptibly(taskManager.getLockingSleepIntervalMillisecond(),
                                              TimeUnit.MILLISECONDS);
      }
      if (System.currentTimeMillis() - begin > taskManager.getLockingMaxMillisecond()) {
        logger.error("作业锁定后等待超时！taskId={}, maxMillis={})", id , taskManager.getLockingMaxMillisecond());
        throw new TaskExecuteException("作业锁定后等待超时！taskId=" + id +
                                         ", maxMillis=" + taskManager.getLockingMaxMillisecond() + ")");
      }
    }
  }


  @Transactional(rollbackFor = Exception.class)
  public SysTask saveOrUpdate(SysTask task){
    return saveAndFlush(task);
  }

  @Transactional
  public SysTask pauseOrStart(String taskId, boolean isPause){
    Assert.isTrue(!Strings.isNullOrEmpty(taskId), "taskId 不能为空");
    SysTask task = findById(taskId);
    Assert.notNull(task, "未找到作业对象, taskId = "+taskId);
    task.setDisabled(isPause);
    saveAndFlush(task);
    return task;
  }

  @Transactional
  public SysTask addTask(Map<String, String> paramMap){
    String taskName = paramMap.get("taskName");
    Assert.isTrue(!Strings.isNullOrEmpty(taskName), "作业名称不能空");
    String targetClass = paramMap.get("targetClass");
    try {
      Class.forName(targetClass);
    } catch (ClassNotFoundException e) {
      throw new IllegalArgumentException("作业执行类未找到："+targetClass);
    }
    String cronExpression = paramMap.get("cronExpression");
    Assert.isTrue(!Strings.isNullOrEmpty(cronExpression), "作业执行周期不能空");
    String argumentsMap = paramMap.get("argumentsMap");
    if(!Strings.isNullOrEmpty(argumentsMap)){
      try {
        JsonUtils.reader(argumentsMap, ObjectNode.class);
      } catch (Exception e) {
        throw new IllegalArgumentException("作业执行调用方法时传递的参数json格式错误");
      }
    }
    SysTask task = newTask();
    task.setId(UUIDUtils.generateKey());
    task.setName(taskName);
    task.setDescription(paramMap.get("description"));
    task.setTargetClass(targetClass);
    task.setCronExpression(cronExpression);
    task.setDisabled(Boolean.FALSE);
    task.setLocking(Boolean.FALSE);
    task.setCreateTime(new Date());
    return saveAndFlush(task);
  }

  /**
   * 修改作业基本信息，手动管理事务
   * @param paramMap
   * @return
   */
  public synchronized SysTask updateTask(Map<String, String> paramMap){
    String taskId = paramMap.get("taskId");
    Assert.hasText(taskId, "作业ID不能空");
    taskManager.waitingBlockingDoingTask(taskId);
    DefaultTransactionDefinition def =
      new DefaultTransactionDefinition(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
    TransactionStatus status = transactionManager.getTransaction(def);
    SysTask task;
    try {
      task = findForUpdate(taskId);
      Assert.notNull(task, "作业任务未找到! id = "+taskId);
      Assert.isTrue(!task.getLocking(), "当前作业正在执行，无法进行修改! id = "+taskId);
      String taskName = paramMap.get("taskName");
      if(StringUtils.isNotBlank(taskName)){
        task.setName(taskName);
      }
      String targetClass = paramMap.get("targetClass");
      if(StringUtils.isNotBlank(targetClass)){
        try {
          Class.forName(targetClass);
          task.setTargetClass(targetClass);
        } catch (ClassNotFoundException e) {
          throw new IllegalArgumentException("作业执行类未找到："+targetClass);
        }
      }
      String cronExpression = paramMap.get("cronExpression");
      if(StringUtils.isNotBlank(cronExpression)){
        task.setCronExpression(cronExpression);
      }
      String argumentsMap = paramMap.get("argumentsMap");
      if(StringUtils.isNotBlank(argumentsMap)){
        try {
          JsonUtils.reader(argumentsMap, ObjectNode.class);
          task.setArgumentsMap(argumentsMap);
        } catch (Exception e) {
          throw new IllegalArgumentException("作业执行调用方法时传递的参数json格式错误");
        }
      }
      String description = paramMap.get("description");
      if(StringUtils.isNotBlank(description)){
        task.setDescription(description);
      }
      getEntityManager().merge(task);
      transactionManager.commit(status);
    } catch (Exception e) {
      transactionManager.rollback(status);
      throw e;
    }
    return task;
  }



  @Transactional(rollbackFor = Exception.class)
  public int unlock(List<String> ids){
    if(ids == null || ids.isEmpty()){
      return 0;
    }
    int count = updateLocking(ids, false);
    Assert.isTrue(count >= 0 && ids.size() == count, "作业解锁失败，需要修改的行数与结果不相等，taskId：" + ids);
    for(String id : ids){
      TaskManager.removeDoingTask(id);
    }
    return count;
  }


  protected abstract SysTask findForUpdate(String id);

  protected abstract EntityManager getEntityManager();

  public abstract SysTask saveAndFlush(SysTask sysTask);

  public abstract SysTask newTask();

  public abstract SysTask findById(String id);

  protected abstract int updateLocking(Collection<String> ids, boolean locking);

  public abstract <T extends SysTask> List<T> findByIds(Iterable<String> ids);

  public abstract <T extends SysTask> Page<T> findAll(ConditionGroup conditionGroup, Pageable pageable);

  public abstract <T extends SysTask> List<T> findEnableTasks();

}